package com.alibaba.bink;

import com.galeno.FlinkUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Tuple;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2817:37
 */
public class 单词统计 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines = env.socketTextStream(FlinkUtils.Linux3, 22222);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordand1 = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] split = value.split("\\s+");
                for (String word : split) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = wordand1.keyBy(x -> x.f0).sum(1);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed2 = wordand1.keyBy(x -> 2).sum(1);
        sumed.print();
        env.execute();


    }
}
